package cn.hloger.spider.demo.webSocketDemo.server;

import cn.hloger.spider.demo.webSocketDemo.bean.RwZxRzVO;
import cn.hloger.spider.demo.webSocketDemo.handle.WsMessageHandle;
import cn.hloger.spider.demo.webSocketDemo.service.SessionHandle;
import cn.hutool.core.thread.NamedThreadFactory;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.PreDestroy;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * socket的客户端
 * onOpen: 当客户端建立 WebSocket 连接时触发。
 * onMessage: 当接收到客户端发送的消息时触发。
 * onClose: 当客户端关闭 WebSocket 连接时触发。
 * onError: 当出现错误时触发。
 *
 * @author huanglong
 * @version 1.0
 * @date 2024/2/27
 */
@Slf4j
public abstract class BaseWebsocketServer {


    // 连接超时 默认2分钟
    public static final long MAX_TIME_OUT = 2 * 60 * 1000;

    // 在线session人数,使用原子类
    public static final AtomicInteger onlineNum = new AtomicInteger();

    public static final ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();

    /**
     * 消息缓存队列池子
     */
    public static final ConcurrentHashMap<String, WsMessageHandle> messageHandlePool = new ConcurrentHashMap<>();


    /**
     * 消息发送线程池 创建一个空闲为0 最长200个的核心消息发送线程池
     */
    public final ExecutorService executorService = new ThreadPoolExecutor(0, 200, 0L, TimeUnit.MILLISECONDS,
            new SynchronousQueue<>(), new NamedThreadFactory("message-handle-", true));


    /**
     * 后续使用缓存队列发送消息 该消息直接作废 发送消息
     *
     * @param session
     * @param message
     */
    public void sendMessage(String message, Session session) {
        if (session != null) {
            log.info(message);
            synchronized (session) {
                try {
                    session.getBasicRemote().sendText(message);
                    log.info("socket发送消息成功");
                } catch (IOException e) {
                    log.error(e.getMessage());
                }
            }
        }
    }

    /**
     * 全员发送
     */
    public void sendAllMessage(String message) {
        for (Session session : sessionPools.values()) {
            sendMessage(message, session);
        }
    }

    /**
     * 全员发送
     */
    public void sendAllWithOutSession(String message, Session session) {
        for (Session localSession : sessionPools.values()) {
            if (!session.getId().equals(localSession.getId())) {
                sendMessage(message, localSession);
            }
        }
    }

    /**
     * 给指定客户端发消息
     *
     * @param message
     */
    public void sendMessage(String message) {
        RwZxRzVO rwZxRzVO = JSON.parseObject(message, RwZxRzVO.class);
        String taskId = rwZxRzVO.getSqlId();
        // 获取当前sqlid的缓存队列
        WsMessageHandle handle = messageHandlePool.computeIfAbsent(taskId, key -> new WsMessageHandle(taskId));
        handle.save(rwZxRzVO);

        // 判断session是否打开，如果已经打开，则推送消息队列中的日志
        Session session = sessionPools.get(rwZxRzVO.getSqlId());
        if (Objects.nonNull(session)) {
            handle.sendMsg(session, executorService);
        }
    }

    /**
     * 连接成功
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session,
                       @PathParam(value = "taskId") String taskId) {
        String name = SessionHandle.getNameBySession(session);
        //1.链接成功,把session放入map中,之后使用的时候方便发送
        sessionPools.put(taskId, session);
        //2. 给session加钟
        session.setMaxIdleTimeout(MAX_TIME_OUT);
        log.info("{}加入webSocket！当前人数为{}", name, onlineNum.incrementAndGet());
        // 查询当前id是否存在队列,如果不在,那么就是新加入的
        WsMessageHandle handle = messageHandlePool.get(taskId);
        if (Objects.nonNull(handle)) {
            handle.sendMsg(session, executorService);
        } else {
            sendMessage(String.format("恭喜%s,连接成功,当前链接人数:%s,", name, onlineNum), session);
        }
    }


    /**
     * 给指定客户端发消息
     *
     * @param message
     */
    @OnMessage
    public abstract void onMessage(String message, Session session) throws IOException;

    @OnError
    public void onError(Throwable t) {
        t.printStackTrace();
    }

    @OnClose
    public void onClose(Session session,
                        @PathParam(value = "taskId") String taskId) {
        String name = SessionHandle.getNameBySession(session);
        sessionPools.remove(taskId);
        log.info("{}断开webSocket连接！当前人数为{}", name, onlineNum.decrementAndGet());
        // 查询当前sqlid是否存在队列,
        WsMessageHandle handle = messageHandlePool.get(taskId);
        if (Objects.nonNull(handle)) {
            handle.destory();
            messageHandlePool.remove(taskId);// 移除该队列
        }
    }


    public @PreDestroy void close() {
        executorService.shutdownNow();
    }
}
